package com.hudson.broker.utils;

import com.hudson.broker.config.BrokerConfig;
import com.hudson.broker.consumer.WaitRequest;
import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.MessageExtBrokerInner;
import com.hudson.store.commitlog.CommitLogManger;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/8 21:36
 **/
public class StoreUtil {


    //旧的protobuf API,暂不删除,先添加新API
//    public static String putMessage(String topic, int queueId, ByteString message) {
//        String msgId;
//
//        if (BrokerConfig.FLUSH_METHOD) {
//            CommitLogManger.asyncAppendMessage(topic, queueId, 0, message);
//            msgId = "";
//        } else {
//            msgId = CommitLogManger.appendMessage(topic, queueId, message);
//        }
//        WaitRequest.notifyMessageArrived(topic, queueId);
//
//        return msgId;
//    }

    //兼容新codec的API
    public static String putMessage(Message message) {
        MessageExtBrokerInner messageExtBrokerInner = MessageExtBrokerInner.builder()
                .topic(message.getTopic())
                .queueId(message.getQueueId())
                .tag(message.getTags())
                .commitLogOffset(0L)
                .bornTimestamp(System.currentTimeMillis())
                .storeTimestamp(System.currentTimeMillis())
                .retryNum(0)
                .body(message.getBody())
                .build();

        String msgId;

        if (BrokerConfig.FLUSH_METHOD) {
            CommitLogManger.asyncAppendMessage(messageExtBrokerInner);
            msgId = "";
        } else {
            msgId = CommitLogManger.appendMessage(messageExtBrokerInner);
        }
        WaitRequest.notifyMessageArrived(message.getTopic(), message.getQueueId());

        return msgId;
    }

    public static void putRetryMessage(MessageExt messageExt) {
        MessageExtBrokerInner messageExtBrokerInner = MessageExtBrokerInner.builder()
                .topic(messageExt.getTopic())
                .queueId(messageExt.getQueueId())
                .tag(messageExt.getTags())
                .commitLogOffset(0L)
                .bornTimestamp(messageExt.getBornTimestamp())
                .storeTimestamp(System.currentTimeMillis())
                .retryNum(messageExt.getReconsumeTimes())
                .body(messageExt.getBody())
                .build();
        if (BrokerConfig.FLUSH_METHOD) {
            CommitLogManger.asyncAppendMessage(messageExtBrokerInner);
        } else {
            CommitLogManger.appendMessage(messageExtBrokerInner);
        }
        WaitRequest.notifyMessageArrived(messageExt.getTopic(), messageExt.getQueueId());
    }

    public static void putDLQMessage(MessageExt messageExt) {
        MessageExtBrokerInner messageExtBrokerInner = MessageExtBrokerInner.builder()
                .topic("DLQ")
                .queueId(0)
                .tag(messageExt.getTags())
                .commitLogOffset(0L)
                .bornTimestamp(messageExt.getBornTimestamp())
                .storeTimestamp(System.currentTimeMillis())
                .retryNum(messageExt.getReconsumeTimes())
                .body(messageExt.getBody())
                .build();
        CommitLogManger.appendMessage(messageExtBrokerInner);

    }
}
